package com.tranq;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.StringTokenizer;

public class TrafficStatMapper extends Mapper<Object, Text, Text, TrafficStatRecord> {
    private final static IntWritable one = new IntWritable(1);
    private final Text outKey = new Text();

    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString());
        ArrayList<Object> fields = Collections.list(itr);
        int size = fields.size();
        int up = Integer.parseInt(fields.get(size-3).toString());
        int down = Integer.parseInt(fields.get(size-2).toString());
        TrafficStatRecord record = new TrafficStatRecord(up, down, up+down);
        outKey.set(fields.get(1).toString());

        context.write(outKey, record);
    }
}
